package com.speily.zookeeperdemo.lock;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @Auther: SPL
 * @Date: 2019-08-12 13:07
 * @Description: - 分布式锁实现demo
 */
@Slf4j
public class LockingExample {
    private static final String CONNECTION_STRING = "47.94.165.79:2181";
    private static final String PATH = "/examples/locks";

    public static void main(String[] args) throws Exception {

        //FakeLimitedResource模拟某些外部资源，这些外部资源一次只能由一个进程访问
        final FakeLimitedResource resource = new FakeLimitedResource();

        ExecutorService service = Executors.newFixedThreadPool(5);//创建一个固定线程池
        try {
            for ( int i = 0; i < 5; ++i ){
                final int index = i;
                Callable<Void> task = () -> {
                    //创建zk客户端
                    CuratorFramework client = CuratorFrameworkFactory.newClient(CONNECTION_STRING, new ExponentialBackoffRetry(1000, 3,Integer.MAX_VALUE));
                    try {
                        client.start();//连接zk
                        //添加ConnectionStateListener监听
                        MyConnectionStateListener connectionStateListener = new MyConnectionStateListener(PATH,"123456");
                        client.getConnectionStateListenable().addListener(connectionStateListener);
                        //资源操作实例
                        ExampleClientThatLocks example = new ExampleClientThatLocks(client, PATH, resource, "Client " + index);
                        for ( int j = 0; j < 10; ++j ) {
                            example.doWork(10, TimeUnit.SECONDS);
                        }
                    }catch ( InterruptedException e ){
                        Thread.currentThread().interrupt();
                    }catch ( Exception e ){
                        e.printStackTrace();
                    }finally{
                        CloseableUtils.closeQuietly(client);
                    }
                    return null;
                };
                service.submit(task);
            }

            service.shutdown();
            service.awaitTermination(10, TimeUnit.MINUTES);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 共享资源（火车票）
     */
    static class FakeLimitedResource {
        //总共50张火车票
        private Integer ticket = 50;

        public void use() throws InterruptedException {
            try {
                log.info("===》火车票还剩"+(--ticket)+"张！");
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }

    /**
     * 资源客户端（操作资源）
     */
    static class ExampleClientThatLocks {

        /** 锁 */
        private final InterProcessMutex lock;
        /** 共享资源 */
        private final FakeLimitedResource resource;
        /** 客户端名称 */
        private final String clientName;

        public ExampleClientThatLocks(CuratorFramework client, String lockPath, FakeLimitedResource resource, String clientName) {
            this.resource = resource;
            this.clientName = clientName;
            lock = new InterProcessMutex(client, lockPath);
        }

        public void doWork(long time, TimeUnit unit) throws Exception {

            //使用acquire方法获取锁，等待time * unit时间获取锁，如果仍然没有获取锁，则直接返回false。
            if ( !lock.acquire(time, unit) ) {
                throw new IllegalStateException(clientName + " could not acquire the lock");
            }
            try {
                log.info(clientName + " has the lock");
                //操作资源
                resource.use();
            } finally {
                log.info(clientName + " releasing the lock");
                lock.release(); //总是在Final块中释放锁。如果该线程多次调用 了acquire()获取锁，则如果只调用 一次release()该锁仍然会被该线程持有。
            }
        }
    }
}